package com.ngkj.zmfptback.sys.service.write.impl;

import com.ngkj.zmfptback.sys.mapper.write.RentalInfoMapper;
import com.ngkj.zmfptback.sys.service.write.RentalInfoService;
import com.ngkj.zmfptCommon.common.bean.RentalInfo;
import com.ngkj.zmfptCommon.common.bean.RpcRequestMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import sun.rmi.runtime.Log;

import javax.annotation.Resource;

/**
 * @author ChenHN
 * @since 2021/11/8 creation
 **/
@Service
@Slf4j
@Transactional(propagation = Propagation.REQUIRED)
public class RentalInfoServiceImpl implements RentalInfoService {

    private static String topicName = "zmfpt-es";

    @Resource
    private RentalInfoMapper rentalInfoMapper;

    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;


    /**
     * 存储租房信息到数据库后，向kafka发送消息。es接收处理
     * @param rentalInfo
     */
    @Override
    public void addRentalInfo(RentalInfo rentalInfo) {
        rentalInfoMapper.addRentalInfo(rentalInfo);
        log.error("已存入数据库.......");
        /**
         * 以rpc方式进行反射调用 存入es
         */
        RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
                "com.ngkj.zmfptes.service.RentalInfoService",
                "saveRentalInfoToEs",
                Void.class,
                new Class[]{RentalInfo.class},
                new Object[]{rentalInfo}
                );
        kafkaTemplate.send(topicName,rpcRequestMessage).addCallback((success)->{ //回调函数
            //消息发送到的主题
            String topic = success.getRecordMetadata().topic();
            //消息发送的分区
            int partition = success.getRecordMetadata().partition();
            log.info("发送消息成功：{} - {} ",topic,partition);
        },(failure)->{
            log.warn("发送消息失败：{}",failure.getMessage());
        });
    }
}
